/*
 * This file is part of Catfish.
 * Copyright (C) 2010 Namazu Studios LLC
 * 
 * Catfish is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as 
 * published by the Free Software Foundation, either version 3 of 
 * the License, or (at your option) any later version.
 * 
 * Catfish is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with Catfish.  If not, please visit:
 *  
 * http://www.gnu.org/licenses/
 *  
 */
package com.namazustudios.catfish.gae;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.ConcurrentModificationException;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.PreparedQuery;
import com.google.appengine.api.datastore.Query;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.api.memcache.MemcacheService;
import com.google.inject.Inject;
import com.google.inject.Injector;
import com.namazustudios.catfish.Catfish;
import com.namazustudios.catfish.CatfishException;
import com.namazustudios.catfish.Configuration;
import com.namazustudios.catfish.DataSource;
import com.namazustudios.catfish.Marshaller;
import com.namazustudios.catfish.StackedDataSource;
import com.namazustudios.catfish.Unmarshaller;
import com.namazustudios.catfish.annotation.Cacheable;
import com.namazustudios.catfish.annotation.EntityKey;
import com.namazustudios.catfish.callback.CriticalSection;
import com.namazustudios.catfish.callback.Delete;
import com.namazustudios.catfish.callback.Get;
import com.namazustudios.catfish.callback.Put;
import com.namazustudios.catfish.exception.CacheInconsistencyException;
import com.namazustudios.catfish.exception.CacheMissException;
import com.namazustudios.catfish.exception.MissingKeyException;
import com.namazustudios.catfish.exception.TransactionTimeoutException;
import com.namazustudios.catfish.generator.IdentityGenerator;
import com.namazustudios.catfish.utility.MethodIterable;

/**
 * Wraps a DatastoreService object and allows you to perform queries on the
 * datastore as well as marshalls entities into objects whose classes contain
 * the proper annotations. This is designed to be a thin layer of abstraction
 * over the Datastore API.
 * 
 * @author patricktwohig
 * 
 */
public class GaeCatfish implements Catfish {

    static Class<com.namazustudios.catfish.annotation.Delete> delete() {
        return com.namazustudios.catfish.annotation.Delete.class;
    }

	static Class<com.namazustudios.catfish.annotation.Entity> entity() {
        return com.namazustudios.catfish.annotation.Entity.class;
    }

    static Class<com.namazustudios.catfish.annotation.Get> get() {
        return com.namazustudios.catfish.annotation.Get.class;
    }

    static Class<com.namazustudios.catfish.annotation.EntityKey> key() {
        return com.namazustudios.catfish.annotation.EntityKey.class;
    }

    static Class<com.namazustudios.catfish.annotation.PostPut> postPut() {
        return com.namazustudios.catfish.annotation.PostPut.class;
    }

    static Class<com.namazustudios.catfish.annotation.PrePut> prePut() {
        return com.namazustudios.catfish.annotation.PrePut.class;
    }

    @Inject
    private Injector injector;

    @Inject
    private MemcacheService memcache;

    @Inject
    private DatastoreService datastore;

    @Inject
    private Configuration configuration;

//    @Inject(optional = true)
//    private Map<Class<?>, Put<?>> puts;
//
//    @Inject(optional = true)
//    private Map<Class<?>, Get<?>> gets;
//
//    @Inject(optional = true)
//    private Map<Class<?>, Delete<?>> deletes;
//
//    @Inject
//    private Map<String, IdentityGenerator<?>> identityGenerators;

    @Inject(optional = true)
    private Logger logger = Logger.getLogger(GaeCatfish.class.getName());

    private int retries = 10;

    @Inject
    private DataSource dataSource;

    private final ThreadLocal<Boolean> cachingEnabled = new ThreadLocal<Boolean>();

    private final ThreadLocal<Deque<StackedDataSource>> sources = new ThreadLocal<Deque<StackedDataSource>>();

    @Override
    public Transaction beginTransaction() {
        return datastore.beginTransaction();
    }

    @Override
	public <T> Map<Key, T> cacheGet(Class<T> cls, Iterable<Key> keys) {
		Map<Key, T> result = new HashMap<Key, T>();
		ArrayList<Object> objects = new ArrayList<Object>();

		if (cls == null) {
			throw new IllegalArgumentException("Class must not be null.");
		} else if (keys == null) {
			throw new IllegalArgumentException("Keys must not be null.");
		}

		for (Key key : keys) {
			objects.add(key);
		}

		for (Entry<Object, Object> entry : memcache.getAll(objects).entrySet()) {
			Key key = (Key)entry.getKey();

			if (cls.isInstance(entry.getValue())) {
				result.put(key, inject(cls.cast(entry.getValue())));
			} else {
				throw new CacheInconsistencyException(key);
			}

		}

		return result;
	}

    @Override
	public <T> Map<Key, T> cacheGet(Class<T> cls, Key... keys) {
		return cacheGet(cls, Arrays.asList(keys));
	}

    @Override
	public <T> T cacheGet(Class<T> cls, Key key) throws CacheMissException {
		Object object;

		if (cls == null) {
			throw new IllegalArgumentException("Class must not be null.");
		} else if (key == null) {
			throw new IllegalArgumentException("Key must not be null.");
		}

		object = memcache.get(key);

		if (object == null) {
			throw new CacheMissException("No object fount for key: " + key);
		} else if (!cls.isInstance(object)) {
			throw new CacheInconsistencyException(key);
		} else {
			return inject(cls.cast(object));
		}
	}

    @Override
	public Map<Key, Object> cacheGet(Iterable<Key> keys) {
		Map<Key, Object> result = new HashMap<Key, Object>();
		ArrayList<Object> objects = new ArrayList<Object>();

		if (keys == null) {
			throw new IllegalArgumentException("Keys must not be null.");
		}

		for (Key key : keys) {
			objects.add(key);
		}

		for (Entry<Object, Object> entry : memcache.getAll(objects).entrySet()) {
			Key key = (Key)entry.getKey();
			result.put(key, inject(entry.getValue()));
		}

		return result;
	}

    @Override
	public Map<Key, Object> cacheGet(Key... keys) {
		return cacheGet(Arrays.asList(keys));
	}

    @Override
	public Object cacheGet(Key key) throws CacheMissException {
		Object object;

		if (key == null) {
			throw new IllegalArgumentException("Key must not be null.");
		}

		object = memcache.get(key);

		if (object == null) {
			throw new CacheMissException("No object fount for key: " + key);
		} else {
			return inject(object);
		}

	}

    @Override
    public void delete(Iterable<? extends Object> objects) {
        Set<Key> keys = new HashSet<Key>();

        for (Object obj : objects) {
            Key key = getKey(obj);

            keys.add(key);
            firePreDelete(obj.getClass(), obj);

            if (isCacheable(obj)) {
                logger.finer("Expiring memcached object: " + key);
            }

        }

        getDataSource().delete(keys);
        memcache.deleteAll(new HashSet<Object>(keys));

    }

    @Override
    public <T> void delete(T... objects) {
        delete(Arrays.asList(objects));
    }

    @Override
    public void delete(Transaction transaction, Iterable<? extends Object> objects) {
        Set<Key> keys = new HashSet<Key>();

        for (Object obj : objects) {
            Key key = getKey(obj);

            keys.add(key);
            firePreDelete(obj.getClass(), obj);

            if (isCacheable(obj)) {
                logger.finer("Expiring memcached object: " + key);
            }

        }

        getDataSource().delete(transaction, keys);
        memcache.deleteAll(new HashSet<Object>(keys));

        for (Key key : keys) {
            logger.finer("Expiring memcache object: " + key);
        }

    }

    @Override
    public <T> void delete(Transaction transaction, T... objects) {
        delete(transaction, Arrays.asList(objects));
    }

    @Override
    public <T> Map<Key, T> get(Class<T> cls, Iterable<Key> keys) {
        Set<Key> keys_ = new HashSet<Key>();
        Map<Key, T> result = new HashMap<Key, T>();

        for (Key k : keys) {
            keys_.add(k);
        }

        if (isCachingEnabled() && isCacheable(cls)) {
            Map<Object, Object> cached;
            cached = memcache.getAll(new HashSet<Object>(keys_));

            try {
                for (Entry<Object, Object> e : cached.entrySet()) {
                    Object value = e.getValue();

                    inject(value);

                    result.put((Key) e.getKey(), cls.cast(value));
                }
            } catch (ClassCastException ex) {
                throw new CatfishException("Types mismatched from " + "memcache service.", ex);
            }

            keys_.removeAll(cached.keySet());
        }

        for (Entry<Key, Entity> e : getDataSource().get(keys_).entrySet()) {
            T obj = unmarshall(cls, e.getValue());

            result.put(e.getKey(), obj);
        }

        return result;
    }

    @Override
    public <T> Map<Key, T> get(Class<T> cls, Key... keys) {
        return get(cls, Arrays.asList(keys));
    }

    @Override
    public <T> T get(Class<T> cls, Key key) throws EntityNotFoundException {
        T result;

        try {
            if (isCachingEnabled() && isCacheable(cls)) {
                result = cls.cast(memcache.get(key));

                if (result != null) {
                    inject(result);
                    logger.finer("Memcache hit for object: " + key);
                } else {
                    logger.finer("Memcache miss for object: " + key);
                }

            } else {
                result = null;
            }
        } catch (ClassCastException ex) {
            throw new CatfishException("Type mismatch with cached value for " + key + " .", ex);
        }

        if (result == null) {
            result = unmarshall(cls, getDataSource().get(key));
        }

        return result;
    }

    @Override
    public <T> Map<Key, T> get(Class<T> cls, Transaction transaction, Iterable<Key> keys) {
        Set<Key> keys_ = new HashSet<Key>();
        Map<Key, T> result = new HashMap<Key, T>();

        for (Key k : keys) {
            keys_.add(k);
        }

        if (isCachingEnabled() && isCacheable(cls)) {
            Map<Object, Object> cached;

            cached = memcache.getAll(new HashSet<Object>(keys_));

            try {
                for (Entry<Object, Object> e : cached.entrySet()) {
                    Object value = e.getValue();

                    inject(value);

                    result.put((Key) e.getKey(), cls.cast(value));
                }
            } catch (ClassCastException ex) {
                throw new CatfishException("Types mismatched from " + "memcache service.", ex);
            }

            keys_.removeAll(cached.keySet());
        }

        for (Entry<Key, Entity> e : getDataSource().get(transaction, keys_).entrySet()) {
            T obj = unmarshall(cls, transaction, e.getValue());

            result.put(e.getKey(), obj);
        }

        return result;
    }

    @Override
    public <T> Map<Key, T> get(Class<T> cls, Transaction transaction, Key... keys) {
        return get(cls, transaction, Arrays.asList(keys));
    }

    @Override
    public <T> T get(Class<T> cls, Transaction transaction, Key key) throws EntityNotFoundException {
        T result;

        try {
            if (isCachingEnabled() && isCacheable(cls)) {
                result = cls.cast(memcache.get(key));

                if (result != null) {
                    inject(result);
                    logger.finer("Memcache hit for object: " + key);
                } else {
                    logger.finer("Memcache miss for object: " + key);
                }

            } else {
                result = null;
            }
        } catch (ClassCastException ex) {
            throw new CatfishException("Type mismatch with cached value for " + key + " .", ex);
        }

        if (result == null) {
            Entity entity = getDataSource().get(transaction, key);
            result = unmarshall(cls, transaction, entity);
        }

        return result;
    }

    @Override
    public Map<Key, Object> get(Iterable<Key> keys) {
        Map<Object, Object> cached;
        Set<Key> keys_ = new HashSet<Key>();
        Map<Key, Object> result = new HashMap<Key, Object>();

        for (Key k : keys) {
            keys_.add(k);
        }

        if (isCachingEnabled()) {
            cached = memcache.getAll(new HashSet<Object>(keys_));

            for (Object value : cached.values()) {
                inject(value);
            }

        } else {
            cached = Collections.emptyMap();
        }

        try {
            for (Entry<Object, Object> e : cached.entrySet()) {
                Key k = (Key) e.getKey();
                Class<?> cls = getClass(k);
                result.put(k, cls.cast(e.getValue()));
            }
        } catch (ClassCastException ex) {
            throw new CatfishException("Types mismatched from " + "memcache service.", ex);
        }

        keys_.removeAll(cached.keySet());

        for (Entry<Key, Entity> e : getDataSource().get(keys_).entrySet()) {
            Class<?> cls = getClass(e.getKey());
            Object obj = unmarshall(cls, e.getValue());
            result.put(e.getKey(), obj);
        }

        return result;
    }

    @Override
    public Map<Key, Object> get(Key... keys) {
        return get(Arrays.asList(keys));
    }

    @Override
    public Object get(Key key) throws EntityNotFoundException {
        boolean cached;
        Object result = (cached = isCachingEnabled()) ? memcache.get(key) : null;

        if (result == null) {
            Entity entity = getDataSource().get(key);
            result = unmarshall(entity);

            if (cached) {
                logger.finer("Memcache miss for object: " + key);
            }

        } else {

            inject(result);

            if (cached) {
                logger.finer("Memcache hit for object: " + key);
            }
        }

        return result;
    }

    @Override
    public Map<Key, Object> get(Transaction transaction, Iterable<Key> keys) {
        Map<Object, Object> cached;
        Set<Key> keys_ = new HashSet<Key>();
        Map<Key, Object> result = new HashMap<Key, Object>();

        for (Key k : keys) {
            keys_.add(k);
        }

        if (isCachingEnabled()) {
            cached = memcache.getAll(new HashSet<Object>(keys_));
        } else {
            cached = Collections.emptyMap();
        }

        try {
            for (Entry<Object, Object> e : cached.entrySet()) {
                Object value;
                Key k = (Key) e.getKey();
                Class<?> cls = getClass(k);

                value = e.getValue();
                inject(value);

                result.put(k, cls.cast(value));
            }
        } catch (ClassCastException ex) {
            throw new CatfishException("Types mismatched from " + "memcache service.", ex);
        }

        keys_.removeAll(cached.keySet());

        for (Entry<Key, Entity> e : getDataSource().get(transaction, keys_).entrySet()) {
            Class<?> cls = getClass(e.getKey());
            Object obj = unmarshall(cls, transaction, e.getValue());

            result.put(e.getKey(), obj);

        }

        return result;
    }

    @Override
    public Object get(Transaction transaction, Key key) throws EntityNotFoundException {
        boolean cached;
        Object result = (cached = isCachingEnabled()) ? memcache.get(key) : null;

        if (result == null) {
            Entity entity = getDataSource().get(transaction, key);
            result = unmarshall(transaction, entity);

            if (cached) {
                logger.finer("Memcache miss for object: " + key);
            }
        } else {
            inject(result);

            if (cached) {
                logger.finer("Memcache hit for object: " + key);
            }
        }

        return result;
    }

    @Override
    public Map<Key, Object> get(Transaction transaction, Key... keys) {
        return get(transaction, Arrays.asList(keys));
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * com.namazustudios.catfish.gae.Catfish#getClass(com.google.appengine.api
     * .datastore.Key)
     */
    @Override
    public Class<?> getClass(Key key) {

        if (key == null) {
            throw new IllegalArgumentException("Specified key must not be null.");
        }

        return configuration.getClass(key.getKind());

    }

    @Override
    public Class<?> getClass(String kind) {

        if (kind == null) {
            throw new IllegalArgumentException("Specified kind must not be null.");
        }

        return configuration.getClass(kind);

    }

    @Override
	public DataSource getDataSource() {
		Deque<StackedDataSource> sources = getSources();
		return sources.isEmpty() ? dataSource : sources.peek();
	}

    @Override
    public DatastoreService getDatastore() {
        return datastore;
    }

    @Override
    public Key getKey(Object object) {
        Class<?> cls;

        if (object == null) {
        	throw new IllegalArgumentException("object must not be null.");
        }

        cls = object.getClass();

        do {
            for (java.lang.reflect.Field f : cls.getDeclaredFields()) {
                boolean accessable = f.isAccessible();
                try {
                    f.setAccessible(true);

                    if (!f.isAnnotationPresent(key())) {
                        continue;
                    }

                    if (f.getType().isAssignableFrom(Key.class)) {
                        return (Key) f.get(object);
                    } else {
                        throw new CatfishException("@Key annotation used with" + " incompatible type.");
                    }

                } catch (IllegalArgumentException ex) {
                    throw new CatfishException("Coult not fetch key value.", ex);
                } catch (IllegalAccessException ex) {
                    throw new CatfishException("Coult not fetch key value.", ex);
                } finally {
                    f.setAccessible(accessable);
                }
            }
        } while ((cls = cls.getSuperclass()) != null);

        throw new MissingKeyException("Object " + object + " is missing @Key annotation");
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.namazustudios.catfish.gae.Catfish#getKind(java.lang.Class)
     */
    @Override
    public String getKind(Class<?> cls) {
    	return Catfish.Utility.getKind(cls);
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.namazustudios.catfish.gae.Catfish#getKind(java.lang.Object)
     */
    @Override
    public String getKind(Object object) {
        return Catfish.Utility.getKind(object);
    }

    @Override
    public <T> T instantiate(Class<T> cls) {
        return injector.getInstance(cls);
    }

    @Override
    public <T> T instantiate(Class<T> cls, Key key) {
    	T object = injector.getInstance(cls);
    	setKey(object, key);
    	return object;
    }

    @Override
    public Object instantiate(Key key) {
        return instantiate(getClass(key), key);
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.namazustudios.catfish.gae.Catfish#isCacheable(java.lang.Class)
     */
    @Override
    public boolean isCacheable(Class<?> cls) {
        return cls.isAnnotationPresent(Cacheable.class) && Serializable.class.isAssignableFrom(cls);
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.namazustudios.catfish.gae.Catfish#isCacheable(java.lang.Object)
     */
    @Override
    public boolean isCacheable(Object object) {
        return isCacheable(object.getClass());
    }

    @Override
    public boolean isCachingEnabled() {

        if (cachingEnabled.get() == null) {
            cachingEnabled.set(true);
        }

        if (cachingEnabled.get()) {
            logger.finer("Caching enabled.");
            return true;
        } else {
            logger.finer("Caching disabled.");
            return false;
        }

    }

    @Override
    public Marshaller marshaller() {
        return injector.getInstance(Marshaller.class);
    }

    @Override
    public Key newKey(Class<?> cls, Key parent, String name) {
        String kind = cls.getAnnotation(entity()).kind();

        if (kind.isEmpty()) {
            kind = cls.getSimpleName();
        }

        return KeyFactory.createKey(parent, kind, name);
    }

    @Override
    public Key newKey(Class<?> cls, String name) {
        String kind = cls.getAnnotation(entity()).kind();

        if (kind.isEmpty()) {
            kind = cls.getSimpleName();
        }

        return KeyFactory.createKey(kind, name);
    }

    @Override
	public void popSource() {
		try {
			StackedDataSource source = getSources().pop();
			source.onPop(this);
		} catch (NoSuchElementException ex) {
			logger.log(Level.WARNING, "Inconsistent data source stack.", ex);
		}
	}

    @Override
	public void pushSource(StackedDataSource source) {

		if (source == null) {
			throw new IllegalArgumentException("Source must not be null.");
		}

		source.onPush(this, getDataSource());
		getSources().push(source);

	}

    /*
     * (non-Javadoc)
     * 
     * @see com.namazustudios.catfish.gae.Catfish#put(java.lang.Iterable)
     */
    @Override
    public <T> List<Key> put(Iterable<? extends T> objects) {
        List<Key> keys = new ArrayList<Key>();
        IdentityHashMap<Object, Entity> entities = new IdentityHashMap<Object, Entity>();

        for (Object obj : objects) {
            entities.putAll(marshall(obj));
        }

        if (entities.isEmpty()) {
            return keys;
        }

        for (Entry<Object, Entity> e : entities.entrySet()) {
            Object obj = e.getKey();
            Key key = getKey(obj);

            firePrePut(e.getKey(), e.getValue());

            if (isCacheable(obj)) {
                logger.finer("Expiring memcache object: " + key);
            }
        }

        keys = getDataSource().put(entities.values());
        memcache.deleteAll(new HashSet<Object>(keys));

        for (Entry<Object, Entity> e : entities.entrySet()) {
            firePostPut(e.getKey(), e.getValue());            
        }

        return keys;
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.namazustudios.catfish.Catfish#put(T[])
     */
    @Override
    public <T> List<Key> put(T... objects) {
        return put(Arrays.asList(objects));
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.namazustudios.catfish.gae.Catfish#put(T)
     */
    @Override
    @SuppressWarnings("unchecked")
    public <T> T put(T object) {
        put(Arrays.asList(object));
        return object;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * com.namazustudios.catfish.gae.Catfish#put(com.google.appengine.api.datastore
     * .Transaction, java.lang.Iterable)
     */
    @Override
    public <T> List<Key> put(Transaction transaction, Iterable<T> objects) {
        List<Key> keys = new ArrayList<Key>();
        IdentityHashMap<Object, Entity> entities = new IdentityHashMap<Object, Entity>();

        for (T obj : objects) {
            entities.putAll(marshall(obj));
        }

        if (entities.isEmpty()) {
            return keys;
        }

        for (Entry<Object, Entity> e : entities.entrySet()) {
            Object obj = e.getKey();
            Key key = getKey(obj);

            firePrePut(e.getKey(), e.getValue());

            if (isCacheable(obj)) {
                logger.finer("Expiring memcache object: " + key);
            }

        }

        keys = getDataSource().put(transaction, entities.values());
        memcache.deleteAll(new HashSet<Object>(keys));

        for (Entry<Object, Entity> e : entities.entrySet()) {
            firePostPut(e.getKey(), e.getValue());
        }

        return keys;
    }

    @Override
    public <T> List<Key> put(Transaction transaction, T... objects) {
        return put(transaction, Arrays.asList(objects));
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> T put(Transaction transaction, T object) {
        put(transaction, Arrays.asList(object));
        return object;
    }

    @Override
    public void recursiveDelete(final Iterable<? extends Object> objects) {
        transaction(new CriticalSection<Void, RuntimeException>() {

            @Override
            public Void call(Transaction transaction) {
                recursiveDelete(transaction, objects);
                return null;
            }

        });
    }

    @Override
    public <T> void recursiveDelete(T... objects) {
        recursiveDelete(Arrays.asList(objects));
    }

    @Override
    public void recursiveDelete(Transaction transaction, Iterable<? extends Object> objects) {
        Map<Key, Object> map = new HashMap<Key, Object>();

        for (Object obj : objects) {
            Key key = getKey(obj);
            map.put(key, obj);
            collect(transaction, key, map);
        }

        for (Object obj : map.values()) {
            firePreDelete(obj.getClass(), obj);
        }

        getDataSource().delete(map.keySet());
        memcache.deleteAll(new HashSet<Object>(map.keySet()));
        
        for (Key key : map.keySet()) {
            logger.finer("Expiring memcache object: " + key);
        }

    }

    @Override
    public <T> void recursiveDelete(Transaction transaction, T... objects) {
        recursiveDelete(transaction, Arrays.asList(objects));
    }

    @Override
    public void setCachingEnabled(boolean cachingEnabled) {
        this.cachingEnabled.set(cachingEnabled);
    }

    @Override
	public void setKey(Object object, Key key) {
        Class<?> cls;

        if (object == null) {
        	throw new IllegalArgumentException("object must not be null.");
        } else if (key == null) {
        	throw new IllegalArgumentException("key must not be null.");
        } else if (!getKind(object).equals(key.getKind())) {
        	throw new IllegalArgumentException("Key mismatch!");
        }

        cls = object.getClass();

        do {
            for (java.lang.reflect.Field f : cls.getDeclaredFields()) {
                boolean accessable = f.isAccessible();
                try {
                    f.setAccessible(true);

                    if (!f.isAnnotationPresent(key())) {
                        continue;
                    }

                    if (f.getType().isAssignableFrom(Key.class)) {
                    	f.set(object, key);
                    	return;
                    } else {
                        throw new CatfishException("@Key annotation used with" + " incompatible type.");
                    }

                } catch (IllegalArgumentException ex) {
                    throw new CatfishException("Coult not fetch key value.", ex);
                } catch (IllegalAccessException ex) {
                    throw new CatfishException("Coult not fetch key value.", ex);
                } finally {
                    f.setAccessible(accessable);
                }
            }
        } while ((cls = cls.getSuperclass()) != null);

        throw new MissingKeyException("Object " + object + " is missing @Key annotation");

	}

    @Override
    public <T, E extends Throwable> T transaction(CriticalSection<T, E> critical) throws E {
        boolean caching = isCachingEnabled();

        injector.injectMembers(critical);

        try {

            setCachingEnabled(false);

            for (int i = 0; i < retries; ++i) {
                Transaction transaction = datastore.beginTransaction();

                try {
                    T ret = critical.call(transaction);
                    transaction.commit();
                    return ret;
                } catch (ConcurrentModificationException ex) {
                    logger.info("Transaction failed.  Trying again. (" + i + "/" + retries + ").");
                    continue;
                } finally {
                    if (transaction.isActive()) {
                        transaction.rollback();
                    }
                }
            }

            throw new TransactionTimeoutException("Exceeded maximum number of " + "transaction attempts. [" + retries + "]");

        } finally {
            setCachingEnabled(caching);
        }
    }

    @Override
    public Unmarshaller unmarshaller() {
        return injector.getInstance(Unmarshaller.class);
    }

    private void collect(Transaction transaction, Key parent, Map<Key, Object> output) {
        Query query;
        PreparedQuery pquery;

        query = new Query().setAncestor(parent);
        pquery = datastore.prepare(transaction, query);

        for (Entity e : pquery.asQueryResultIterable()) {
            Key key = e.getKey();
            output.put(key, unmarshall(e));
        }

    }

    private void fireCallback(Object object, Method method) {
        Class<?>[] params = method.getParameterTypes();

        method.setAccessible(true);

        try {
            if (params.length == 0) {
                method.invoke(object);
            } else if (params.length == 1 && params[0].isAssignableFrom(GaeCatfish.class)) {
                method.invoke(object, this);
            } else {
                logger.log(Level.WARNING, "Could not invoke callback: " + method + ".");
            }
        } catch (InvocationTargetException ex) {
            throw new CatfishException("Failed to invoke callback method " + method + ".", ex.getCause());
        } catch (IllegalArgumentException ex) {
            throw new CatfishException(ex);
        } catch (IllegalAccessException ex) {
            throw new CatfishException(ex);
        }

    }

    private void fireCallback(Object object, Method method, Entity entity) {
        Class<?>[] params = method.getParameterTypes();

        method.setAccessible(true);

        try {
            if (params.length == 0) {
                method.invoke(object);
            } else {
                Object[] objects = new Object[params.length];

                for (int i = 0; i < params.length; ++i) {
                    if (params[i].isAssignableFrom(Entity.class)) {
                        objects[i] = entity;
                    } else if (params[i].isAssignableFrom(GaeCatfish.class)) {
                        objects[i] = this;
                    } else {
                        logger.log(Level.WARNING, "Could not invoke callback: " + method + ".");
                        return;
                    }
                }

                if (objects.length == 1) {
                    method.invoke(object, objects[0]);
                } else if (objects.length == 2) {
                    method.invoke(object, objects[0], objects[1]);
                } else {
                    logger.log(Level.WARNING, "Could not invoke callback: " + method + ".");
                }

            }
        } catch (InvocationTargetException ex) {
            throw new CatfishException("Failed to invoke callback method " + method + ".", ex.getTargetException());
        } catch (IllegalArgumentException ex) {
            throw new CatfishException(ex);
        } catch (IllegalAccessException ex) {
            throw new CatfishException(ex);
        }

    }

    @SuppressWarnings("unchecked")
    private <T> void firePostPut(Class<? extends T> cls, T object, Entity entity) {
        Put<T> put;

        for (Method method : new MethodIterable(cls)) {
            if (method.isAnnotationPresent(postPut())) {
                boolean accessible = method.isAccessible();

                try {
                    fireCallback(object, method, entity);
                } finally {
                    method.setAccessible(accessible);
                }

            }
        }

        put = (Put<T>) configuration.getPutCallback(cls);

        if (put != null) {
            put.postPut(this, object);
        }

    }

    private <T> void firePostPut(T object, Entity entity) {
        firePostPut(object.getClass(), object, entity);
    }

    @SuppressWarnings("unchecked")
    private <T> void firePreDelete(Class<? extends T> cls, T object) {
        Delete<T> delete;

        for (Method method : new MethodIterable(cls)) {
            if (method.isAnnotationPresent(delete())) {
                boolean accessible = method.isAccessible();

                try {
                    fireCallback(object, method);
                } finally {
                    method.setAccessible(accessible);
                }

            }
        }

        delete = (Delete<T>) configuration.getDeleteCallback(cls);

        if (delete != null) {
            delete.preDelete(this, object);
        }

    }

    @SuppressWarnings("unchecked")
    private <T> void firePrePut(Class<? extends T> cls, T object, Entity entity) {
        Put<T> put;

        for (Method method : new MethodIterable(cls)) {
            if (method.isAnnotationPresent(prePut())) {
                boolean accessible = method.isAccessible();

                try {
                    fireCallback(object, method, entity);
                } finally {
                    method.setAccessible(accessible);
                }

            }
        }

        put = (Put<T>) configuration.getPutCallback(cls);

        if (put != null) {
            put.prePut(this, object);
        }

    }

    private <T> void firePrePut(T object, Entity entity) {
        firePrePut(object.getClass(), object, entity);
    }

    private Deque<StackedDataSource> getSources() {
    	Deque<StackedDataSource> sources = this.sources.get();

    	if (sources == null) {
    		sources = new LinkedList<StackedDataSource>();
    		this.sources.set(sources);
    	}

    	return sources;
    }

    private IdentityHashMap<Object, Entity> marshall(Object object) {
        return marshaller().marshall(object);
    }

	private <T> T unmarshall(Class<T> cls, Entity entity) {
        Object ret = unmarshall(entity);

        if (cls.isInstance(ret)) {
            return cls.cast(ret);
        } else {
            throw new CatfishException("Unmarshalled instance of " + ret.getClass() + " but expected " + cls);
        }

    }

	private <T> T unmarshall(Class<T> cls, Transaction transaction, Entity entity) {
        Object ret = unmarshall(transaction, entity);

        if (cls.isInstance(ret)) {
            return cls.cast(ret);
        } else {
            throw new CatfishException("Unmarshalled instance of " + ret.getClass() + " but expected " + cls);
        }

    }

	private Object unmarshall(Entity entity) {
        return unmarshaller().unmarshall(entity);
    }

	private Object unmarshall(Transaction transaction, Entity entity) {
        return unmarshaller().unmarshall(transaction, entity);
    }

	@SuppressWarnings("unchecked")
    <T> void firePostGet(Class<? extends T> cls, T object, Entity entity) {
        Get<T> get;

        for (Method method : new MethodIterable(cls)) {
            if (method.isAnnotationPresent(get())) {
                boolean accessible = method.isAccessible();

                try {
                    fireCallback(object, method, entity);
                } finally {
                    method.setAccessible(accessible);
                }

            }
        }

        get = (Get<T>) configuration.getGetCallback(cls);

        if (get != null) {
            get.postGet(this, object);
        }

    }

	@SuppressWarnings("unchecked")
    <T> Key generateKey(T object, Key parent) {
        Key k = getKey(object);
        String kind = getKind(object);
        Class<?> cls = object.getClass();

        if (k != null) {

            if (parent != null && !parent.equals(k.getParent())) {
                throw new CatfishException("Parent and child key mismatch.  " + "Expected: " + parent + ". Got: "
                        + k.getParent() + ".");
            }

            return k;
        }

        do {
            for (Field f : cls.getDeclaredFields()) {
                EntityKey key;
                IdentityGenerator<T> gen;
                boolean accessable = f.isAccessible();

                f.setAccessible(true);

                if (f.isAnnotationPresent(key())) {
                    key = f.getAnnotation(EntityKey.class);
                } else {
                	continue;
                }

                try {
                    String name = f.getAnnotation(key()).identityGenerator();

                    gen = (IdentityGenerator<T>) configuration.getIdentityGenerator(key);

                    if (gen == null) {
                        throw new CatfishException("Identity generator " + name + " not found.");
                    }

                    if (parent == null) {
                        k = gen.generateId(object);
                    } else {
                        k = gen.generateId(parent, object);
                    }

                    if (k == null) {
                        throw new CatfishException("ID Generator " + name + " returned null value.");
                    } else if (!kind.equals(k.getKind())) {
                        throw new CatfishException("ID Generator " + name + " returned incorrect kind.");
                    }

                    f.set(object, k);

                    break;
                } catch (ClassCastException ex) {
                    throw new CatfishException("Could not generate" + " ID for object." + object, ex);
                } catch (IllegalArgumentException ex) {
                    throw new CatfishException("Could not generate" + " ID for object." + object, ex);
                } catch (IllegalAccessException ex) {
                    throw new CatfishException("Could not generate" + " ID for object." + object, ex);
                } finally {
                    f.setAccessible(accessable);
                }
            }
        } while ((cls = cls.getSuperclass()) != null);

        return k;
    }

    <T> T inject(T t) {

        try {
            logger.finest("Injecting members of object: " + getKey(t));            
        } catch (MissingKeyException ex) {
            logger.finest("Injecting members of object: " + t);
        }

        injector.injectMembers(t);

        return t;
    }

}